package com.atguigu.gmall.realtime.app.dwd.db;

import com.atguigu.gmall.realtime.app.BaseSQLApp;
import com.atguigu.gmall.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lzc
 * @Date 2022/5/2 8:46
 */
public class DWD_16_DwdUserRegisterApp extends BaseSQLApp {
    public static void main(String[] args) {
        new DWD_16_DwdUserRegisterApp()
            .init("DwdUserRegisterApp", 3015, 1, "DwdUserRegisterApp");
        
    }
    
    @Override
    public void handle(StreamExecutionEnvironment env,
                       StreamTableEnvironment tEnv) {
        /*
           从 ods_db 中读取用户表数据
            
            注册的时候, 会插入一条数据, 过滤出 insert 数据即可
            
         */
        tEnv.executeSql(
            "create table topic_db(" +
                "`database` string, " +
                "`table` string, " +
                "`type` string, " +
                "`data` map<string, string>, " +
                "`ts` string " +
                ")" + SQLUtil.getKafkaDDL("dwd_trade_order_detail", "ods_db"));
        
        // TODO 4. 读取用户表数据
        Table userInfo = tEnv.sqlQuery(
            "select " +
                "data['id'] user_id, " +
                "data['create_time'] create_time, " +
                "ts " +
                "from topic_db " +
                "where `table` = 'user_info' " +
                "and `type` = 'insert' ");
        tEnv.createTemporaryView("user_info", userInfo);
        
        // TODO 5. 创建 Upsert-Kafka dwd_user_register 表
        tEnv.executeSql(
            "create table `dwd_user_register`( " +
                "`user_id` string, " +
                "`date_id` string, " +
                "`create_time` string, " +
                "`ts` string, " +
                "primary key(`user_id`) not enforced " +
                ")" + SQLUtil.getUpsertKafkaDDL("dwd_user_register"));
        
        // TODO 6. 将输入写入 Upsert-Kafka 表
        tEnv.executeSql(
            "insert into dwd_user_register " +
                "select  " +
                "user_id, " +
                "date_format(create_time, 'yyyy-MM-dd') date_id, " +
                "create_time, " +
                "ts " +
                "from user_info");
    }
}
